package com.atguigu.mr;

import com.atguigu.utils.MySQLOutputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;

import java.io.IOException;

/**
 * Project: DXXM
 * Package: com.atguigu.mr
 * Version: 1.0
 * <p>
 * Created by  WangJX  on 2020/2/21 13:11
 */
public class Start {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://hadoop105:9000");
        conf.set("hbase.zookeeper.quorum", "hadoop105,hadoop106,hadoop107");
        conf.set("hbase.zookeeper.property.clientPort","2181");
        try {

            Job job = Job.getInstance(conf, "MyWordCount");

            job.setJarByClass(Start.class);

            Start.setHBaseInputConfig(job);


            //job.setMapperClass(CountDurationMapper.class);
            job.setReducerClass(CountDurationReducer.class);
            job.setOutputKeyClass(ComDimension.class);
            job.setOutputValueClass(CountDurationValue.class);

            //为Job设置OutputFormat
            job.setOutputFormatClass(MySQLOutputFormat.class);

            boolean b = job.waitForCompletion(true);

            System.exit(b ? 0 : 1);

        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }

    static private void setHBaseInputConfig(Job job) {
        //获取配置文件信息
        Configuration conf = job.getConfiguration();
        HBaseAdmin admin = null;
        try {
            admin = new HBaseAdmin(conf);
            //如果表不存在则直接返回，抛个异常也挺好
            if (!admin.tableExists("ns_telecom:calllog")) {
                throw new RuntimeException("Unable to find the specified table.");
            }

            //配置需要扫描的Hbase表
            Scan scan = new Scan();
            scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes("ns_telecom:calllog"));

            TableMapReduceUtil.initTableMapperJob("ns_telecom:calllog", scan,
                    CountDurationMapper.class, ComDimension.class, Text.class,
                    job, true);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (admin != null) {
                try {
                    admin.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
